package com.smartstate.iot.mqtt.callback;


import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.smartstate.iot.mqtt.client.MyMQTTClient;
import com.smartstate.iot.service.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;


import java.util.*;
import java.util.concurrent.*;

@Slf4j
public class MyMQTTCallback implements MqttCallback {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyMQTTCallback.class);

    @Value("${mqtt.subTopic1}")
    private String subTopic1;

    private MyMQTTClient myMQTTClient;


    private static final ThreadPoolExecutor threadpool = new ThreadPoolExecutor(100, 200,
            10L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());

    public MyMQTTCallback(MyMQTTClient myMQTTClient) {
        this.myMQTTClient = myMQTTClient;
    }

    /**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        try {
            myMQTTClient.connect();
            myMQTTClient.subscribe(subTopic1 + "/#", 1);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param toptic
     * @param message
     * @throws Exception subscribe后得到的消息会执行到这里面
     */
    @Override
    public void messageArrived(String toptic, MqttMessage message) throws Exception {
        log.info("---{}--messageArrived--start---toptic:{}-- ", Thread.currentThread().getName(), toptic);
        if(StringUtils.isBlank(toptic) || message.getPayload() == null || message.getPayload().length == 0){
            log.error("消费消息主题：{}信息不完整！", toptic);
            return;
        }
        String msg = new String(message.getPayload(), "GBK");

        //  网关采集的信息储存在redis
        if(toptic.endsWith("rtg")){
            threadpool.execute(new ThreadService(toptic, msg));
        }
        log.info("---{}--messageArrived--end---{} ", Thread.currentThread().getName(), toptic);
//        //  采集盒子
//        if (topic.contains("ALL")) {
//            if (topic.endsWith("HEART")) {   //  采集盒子心跳
//                TopicUpVo topicUpVo = JSONObject.parseObject(msg, TopicUpVo.class);
//                TopicTool bean = SpringBeanUtils.getBean(TopicTool.class);
//                Boolean Boolean = bean.saveTopicConfig(topicUpVo);
//            } else if (topic.endsWith("CONFIG")) {
//                //  iot下发给盒子，消息过滤，不做处理
//            } else if (topic.endsWith("CONFIGS")) {   //  中台下发盒子报警配置
//                threadpool.execute(new TransmitThreadService("SH/HK/NJ/CONFIG", msg));
//            } else {     //  盒子采集数据
//                //  盒子采集的信息储存在redis
//                threadpool.execute(new
//                (topic, msg));
//                Map map = JSON.parseObject(msg, Map.class);
//                List<TopicUpVo.ConfigDetails> config = (List<TopicUpVo.ConfigDetails>) map.get("config");
//                String newMsg = config.toString();
//                myMQTTClient.publish(newMsg,"BOX/CONFIG/UP/ZT");
//            }
//        }
//        else if (topic.contains("PLC")) {   //采集PLC
//            if (topic.endsWith("HEART")) {   //  采集盒子心跳
//                Map<String, Object> map = JSON.parseObject(msg, Map.class);
//                TopicUpVo beatVO = new TopicUpVo();
//                beatVO.setConfig(map.get("config").toString());
//                beatVO.setTimesTamp(map.get("timesTamp").toString());
//                ArrayList<TopicUpVo.Content> contents = new ArrayList<>();
//                TopicUpVo.Content content1 = new TopicUpVo.Content();
//                Map<String,Object> content = (Map) map.get("content");
//                content1.setDeviceId(content.get("boxId").toString());
//                content1.setDeviceName(content.get("boxName").toString());
//                content1.setDeviceType(content.get("boxType").toString());
//                content1.setIp(content.get("ip").toString());
//                content1.setPort(content.get("port").toString());
//                String cpuTemp = content.get("cpu").toString();
//                String cpu = cpuTemp.substring(0,cpuTemp.length()-1);
//                String memoryTemp = content.get("memory").toString();
//                String memory = memoryTemp.substring(0,memoryTemp.length()-1);
//                content1.setCpu(cpu);
//                content1.setMomery(memory);
//                content1.setNewTopic(content.get("newTopic").toString());
//                contents.add(content1);
//                beatVO.setContent(contents);
//                TopicTool bean = SpringBeanUtils.getBean(TopicTool.class);
//                //  将心跳信息储存
//                Boolean Boolean = bean.saveTopicConfig(beatVO);
//            } else if (topic.endsWith("CONFIG")) {
//                //  iot下发给盒子，消息过滤，不做处理
//            } else if (topic.endsWith("CONFIGS")) {   //  中台下发盒子报警配置
//                threadpool.execute(new PointsAwardedThreadService("SH/HK/NJ/PLC/CONFIG", msg));
//            } else {     //  盒子采集数据
//                threadpool.execute(new PlcThreadService(topic, msg));
//            }
//        }
//        log.info("接收消息主题 : {}，接收消息内容 : {}", toptic, msg);
    }

    /**
     * Json格式转译
     *
     * @param jsonObj
     * @return
     */
    public Map<String, Object> analysis(JSONObject jsonObj) {
        JSONArray readResults = jsonObj.getJSONArray("values");
        Map<String, Object> map = new HashMap<>();
        for (int i = 0; i < readResults.size(); i++) {
            JSONObject jsonObject = readResults.getJSONObject(i);
            String[] split = jsonObject.getString("id").split("\\.");
            String id = split[split.length - 1];
            Object value = 0;
            if (Boolean.parseBoolean(jsonObject.get("q").toString())) {
                value = jsonObject.get("v");
            }
            map.put(id, value);
        }
        return map;
    }

    /**
     * 消息到达后
     * subscribe后，执行的回调函数
     *
     * @param s
     * @param mqttMessage
     * @throws Exception
     */

    /**
     * publish后，配送完成后回调的方法
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
//        LOGGER.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());
    }
}
